package com.pie4cloud.pie.job.biz.service.impl;


import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.pie4cloud.pie.jpa.BaseDao;
import com.pie4cloud.pie.jpa.BaseService;
import com.pie4cloud.pie.jpa.page.PageBean;
import com.pie4cloud.pie.jpa.page.PageForm;
import com.pie4cloud.pie.jpa.specification.SimpleSpecificationBuilder;
import com.pie4cloud.pie.jpa.specification.SpecificationOperator;
import com.pie4cloud.pie.bus.api.entity.DataKafConsumer;
import com.pie4cloud.pie.bus.api.feign.RemoteKafkaManageService;
import com.pie4cloud.pie.common.core.constant.SecurityConstants;
import com.pie4cloud.pie.common.core.util.R;
import com.pie4cloud.pie.job.api.entity.GatherEntity;
import com.pie4cloud.pie.job.biz.dao.GatherDao;
import com.pie4cloud.pie.job.biz.service.GatherService;
import com.pie4cloud.pie.job.biz.service.JobInfoService;
import lombok.RequiredArgsConstructor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Sort;
import org.springframework.data.jpa.domain.Specification;
import org.springframework.stereotype.Service;

import java.util.List;


@Service
@RequiredArgsConstructor
public class GatherServiceImpl extends BaseService<GatherEntity,Long> implements GatherService {

    @Autowired
    private GatherDao gatherDao;
	@Override
	public BaseDao<GatherEntity,Long> getBaseDao() {
		return gatherDao;
	}
	@Autowired
	private JobInfoService jobInfoService;
	private final RemoteKafkaManageService remoteKafkaManageService;

    @Override
    public GatherEntity getById(Long id) {
        return gatherDao.findById(id).get();
    }

	@Override
	public List<GatherEntity> list() {
		return gatherDao.findAll();
	}

	public PageBean<GatherEntity> page(Page page, GatherEntity gatherEntity){
		SimpleSpecificationBuilder specificationBuilder = new SimpleSpecificationBuilder();
		if (null != gatherEntity.getTriggerStatus()) {
			specificationBuilder.add("triggerStatus", SpecificationOperator.Operator.eq.name(), gatherEntity.getTriggerStatus());
		}
		Specification specification = specificationBuilder.generateSpecification();
		Sort sort = Sort.by(Sort.Direction.DESC, "createTime");
		PageForm pageForm=new PageForm((int)page.getCurrent(),(int)page.getSize());
		PageBean pageBean = this.getPage(specification, pageForm, sort);
		return pageBean;
	}

	public GatherEntity save(GatherEntity gatherEntity){

		gatherEntity= gatherDao.saveNotNull(gatherEntity);
		try {
			DataKafConsumer dataKafConsumer=new DataKafConsumer();
			dataKafConsumer.setWorkNum(4);
			dataKafConsumer.setTopic("GATHER_1498950413673615360");
			dataKafConsumer.setKafGroup("GATHER");
			dataKafConsumer.setHandler("gatherKakHandler");
			dataKafConsumer.setServiceName("pie-job-executor");
			dataKafConsumer.setTriggerStatus(gatherEntity.getTriggerStatus());
			R<DataKafConsumer> dataKafConsumerR= remoteKafkaManageService.save(dataKafConsumer, SecurityConstants.FROM_IN);
			if(0==dataKafConsumerR.getCode()){
				gatherEntity.setConsId(dataKafConsumerR.getData().getConsId());
				gatherEntity=gatherDao.saveNotNull(gatherEntity);
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
		jobInfoService.start(gatherEntity);
		return gatherEntity;
	}

	public GatherEntity update(GatherEntity gatherEntity){

		gatherEntity= gatherDao.saveNotNull(gatherEntity);
		try {
			DataKafConsumer dataKafConsumer=new DataKafConsumer();
			dataKafConsumer.setWorkNum(4);
			dataKafConsumer.setTopic("GATHER_1498950413673615360");
			dataKafConsumer.setKafGroup("GATHER");
			dataKafConsumer.setHandler("gatherKakHandler");
			dataKafConsumer.setServiceName("pie-job-executor");
			dataKafConsumer.setTriggerStatus(gatherEntity.getTriggerStatus());
			dataKafConsumer.setConsId(gatherEntity.getConsId());
			R<DataKafConsumer> dataKafConsumerR= remoteKafkaManageService.updateById(dataKafConsumer, SecurityConstants.FROM_IN);
			if(0==dataKafConsumerR.getCode()){
				gatherEntity.setConsId(dataKafConsumerR.getData().getConsId());
				gatherEntity=gatherDao.saveNotNull(gatherEntity);
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
		jobInfoService.start(gatherEntity);
		return gatherEntity;
	}

	public Boolean removeById(Long id){
		GatherEntity dispenseEntity= gatherDao.findById(id).get();
		try {
			remoteKafkaManageService.removeById(dispenseEntity.getConsId(), SecurityConstants.FROM_IN);
		} catch (Exception e) {
			e.printStackTrace();
		}
		gatherDao.deleteById(id);
		 jobInfoService.stop(id);
		 return true;
	}

}

